package com.zhangxiaofanfan.learn.producer;

import com.zhangxiaofanfan.learn.utils.KafkaUtils;
import com.zhangxiaofanfan.learn.pojo.Company;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

/**
 * 生产者: 向 kafka 中发送 Company 对象
 *
 * @author zhangxiaofanfan
 * @date 2023-10-28 21:56:18
 */
public class CompanyProducerStart {
    public static void main(String[] args) {
        // 配置需要连接的 kafka 节点信息
        Properties properties = KafkaUtils.getCompanyProducerProperties();
        // 配置生产者客户端工具参数并创建 KafkaProducer 实例
        KafkaProducer<String, Company> producer = new KafkaProducer<>(properties);
        Company company = Company.builder().name("张小帆帆科技有限公司").address("保定市").build();
        // 构建所需要发送的消息
        ProducerRecord<String, Company> record = new ProducerRecord<>(KafkaUtils.topic, company);
        System.out.println("send msg begin......");
        // 发送消息
        try {
            producer.send(record);
        } catch (Exception e) {
            System.out.printf("error msg is %s", e.getMessage());
        }
        System.out.println("send msg end......");
        // 关闭生产者客户端示例
        producer.close();
    }
}
